// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <cstdlib>
#include <deque>
#include <functional>
#include <iostream>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <tuple>
#include <vector>
#include <optional>

#include "coin2/base/config/app_config.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/factory/feed/request_generator_factory.h"
#include "coin2/exchange/util/rate_limit.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/flow/symbol_group.h"

#include "cc/coin2/exchange/base/feed/ws/rest_connection.h"

#include <fmt/core.h>

namespace coin2::exchange::base::feed {

using coin2::exchange::feed::WebsocketConfig;

class RestSubscriber : public std::enable_shared_from_this<RestSubscriber>,
                     public api_base::BaseRawAsyncSubscriber {
 public:
  RestSubscriber(
      const FeedSubscriptionSpec& spec,
      const WebsocketConfig& config,
      const std::vector<WsConnectionConfig>& connection_configs,
      api_base::LiveAsyncExecutor* flow_sub)
      : BaseRawAsyncSubscriber(flow_sub),
        spec_(spec),
        connection_configs_(connection_configs),
        request_generator_(factory::feed::NewRequestGenerator(spec.mea, &config)),
        strand_(*ioc()),
        timer_(*ioc()),
        subscribe_one_time_(config.subscribe_one_time()) {
  }

  void RegisterPubTopic() override { flow_sub_->distributor()->Publish(spec_.mea.SnakeString()); }

  void SubscribeAsync() override {
    boost::asio::spawn(strand_, std::bind(
        &RestSubscriber::Subscribe, shared_from_this(), std::placeholders::_1));
  }

  void Subscribe(boost::asio::yield_context yield) {
    std::vector<std::string> requests;
    std::vector<std::set<std::string>> topics;
    std::vector<std::vector<std::pair<std::string, std::string>>> headers;
    std::vector<std::string> payloads;
    std::vector<boost::beast::http::verb> methods;
    std::string http_string = "https://";

    // url has the same configuration, only channels dierent
    int64_t num = connection_configs_.front().url.unauthorized_limit.subscription->first;
    int64_t window = connection_configs_.front().url.unauthorized_limit.subscription->second;
    int64_t interval = window / num + 1;
    std::string host = connection_configs_.front().url.host;
    std::string port = connection_configs_.front().url.port;
    std::string bind_ip = connection_configs_.front().url.bind_ip;
    std::string http_proxy = connection_configs_.front().url.proxy;
    bool wrap_response = connection_configs_.front().url.wrap_response;
    BuildRequests(&requests, &topics, &headers, &payloads, &methods);

    boost::system::error_code ec;
    std::shared_ptr<RestConnection> connection = std::make_shared<RestConnection>(
        ioc(),
        host,
        std::stoi(port),
        bind_ip,
        http_proxy);
    connection->Start(yield[ec]);

    auto iter_requests = requests.begin();
    auto iter_headers = headers.begin();
    auto iter_topics = topics.begin();
    auto iter_payloads = payloads.begin();
    auto iter_methods = methods.begin();
    while (sub_running_) {
      if (iter_requests == requests.end()) {
        if (subscribe_one_time_) {
          LOG(INFO) << spec_.mea.String() << " subscribe done.";
          this->StopAsync();
          break;
        }

        // as rest api is a query, parameter may changed in every query
        BuildRequests(&requests, &topics, &headers, &payloads, &methods);
        iter_requests = requests.begin();
        iter_headers = headers.begin();
        iter_topics = topics.begin();
        iter_payloads = payloads.begin();
        iter_methods = methods.begin();
      }
      const std::string &target = *iter_requests;
      const std::vector<std::pair<std::string, std::string>> &header = *iter_headers;
      const std::set<std::string> &topics = *iter_topics;
      const std::string &payload = *iter_payloads;
      std::string response;
      bool status = connection->Send(target, header, *iter_methods, payload, &response, yield[ec]);
      if (status) {
        if (wrap_response) {
          response = WrapResponse(response, host, target, payload);
        }
        for (const std::string &topic : topics) {
          auto packet = std::shared_ptr<executor::TopicRecordPacket>(
              new executor::TopicRecordPacket{
                  topic,
                  0,
                  executor::TopicRecordProcessor::GetTimestamp(),
                  0,
                  0,
                  0,
                  0,
                  0,
                  0,
                  response});
          flow_sub_->injector().push(packet);
        }
      } else {
        if (!response.empty()) {
          LOG_FIRST_N(ERROR, 1000) << "Failed get: " << host<< ":" << port << target <<
              ", response: " << response;
        }
        // the operation may be canceled
        if (!sub_running_) break;
      }

      iter_requests++;
      iter_headers++;
      iter_topics++;
      iter_payloads++;
      iter_methods++;
      timer_.expires_after(std::chrono::nanoseconds(interval));
      timer_.async_wait(yield[ec]);
    }
  }

  void GetHealthReport(int indent, HealthReport* health) const override {
    for (const WsConnectionConfig& config : connection_configs_) {
      for (int ind = 0; ind < indent * 4; ind++) {
        health->report += ' ';
      }
      health->report += config.topic + " " + config.url.name + " (https)\n";
    }
  }

  void CleanUp() override {
    sub_running_ = false;
  }

  bool RemoveSymbolFromConfig(const std::vector<std::string>& symbols) {
    for (auto iter_config = connection_configs_.begin(); iter_config != connection_configs_.end(); ) {
      for (auto iter_symbol = iter_config->symbols.begin(); iter_symbol != iter_config->symbols.end(); ) {
        auto pos = std::find(symbols.begin(), symbols.end(), *iter_symbol);
        if (pos != symbols.end()) {
          iter_symbol = iter_config->symbols.erase(pos);
        } else {
          ++ iter_symbol;
        }
      }
      if (iter_config->symbols.empty()) {
        iter_config = connection_configs_.erase(iter_config);
      } else {
        ++iter_config;
      }
    }

    if (connection_configs_.empty()) {
      CleanUp();
    }

    return true;
  }

  bool BanProduct(const std::vector<const IProduct*>& products) override {
    std::weak_ptr<RestSubscriber> self(shared_from_this());
    std::vector<std::string> removed_symbols;
    for (const auto& product : products) {
      removed_symbols.push_back(product->feed_native_symbol());
    }
    boost::asio::post(
        strand_,
        [self, removed_symbols] {
          if (auto s = self.lock()) {
            s->RemoveSymbolFromConfig(removed_symbols);
          }
        });
    return true;
  }

 private:
  std::string WrapResponse(
      const std::string& data,
      const std::string& host,
      const std::string& target,
      const std::string& payload) {
    if (payload.empty()) {
      return fmt::format("{{\"url\":\"{}{}\", \"data\":{}}}", host, target, data);
    } else {
      return fmt::format("{{\"url\":\"{}{}\", \"payload\":{}, \"data\":{}}}",
          host, target, payload, data);
    }
  }

  void BuildRequests(
      std::vector<std::string>* requests,
      std::vector<std::set<std::string>>* topics,
      std::vector<std::vector<std::pair<std::string, std::string>>>* headers,
      std::vector<std::string> *payloads,
      std::vector<boost::beast::http::verb> *methods ) {
    requests->clear();
    topics->clear();
    headers->clear();
    payloads->clear();
    methods->clear();
    for (const WsConnectionConfig& config : connection_configs_) {
      std::vector<std::string> reqs = request_generator_->GenerateRequests(
          config.url.channels,
          config.symbols,
          config.mea_config->channels,
          config.url.request);
      auto header = request_generator_->GenerateHeader(
          config.url.channels,
          config.symbols,
          config.mea_config->channels,
          config.url.request);
      // merge request as they use same site, and site may has query limitation
      for (const std::string& req : reqs) {
        std::string target;
        std::string payload;
        if (config.url.method == "post") {
          target = config.url.path;
          payload = req;
          // not implement remove duplicate post
          requests->emplace_back(target);
          methods->emplace_back(boost::beast::http::verb::post);
          payloads->emplace_back(payload);
          headers->emplace_back(header);
          std::set<std::string> request_topics;
          request_topics.insert(config.topic);
          topics->emplace_back(request_topics);
        } else {
          target = config.url.path + req;
          auto iter = std::find(requests->begin(), requests->end(), target);
          if (iter == requests->end()) {
            requests->emplace_back(target);
            methods->emplace_back(boost::beast::http::verb::get);
            payloads->emplace_back(payload);
            headers->emplace_back(header);
            std::set<std::string> request_topics;
            request_topics.insert(config.topic);
            topics->emplace_back(request_topics);
          } else {
            int idx = std::distance(requests->begin(), iter);
            topics->at(idx).insert(config.topic);
          }
        }
      }
    }
    CHECK_GT(requests->size(), 0) << "No REST subscribe";
  }
  // connection per symbol group config
  const FeedSubscriptionSpec& spec_;
  std::vector<WsConnectionConfig> connection_configs_;
  std::unique_ptr<RequestGenerator> request_generator_;
  boost::asio::io_context::strand strand_;
  boost::asio::high_resolution_timer timer_;
  bool subscribe_one_time_;
};

}  // namespace coin2::exchange::base::feed
